package com.atguigu.edu.realtime.util;

import com.atguigu.edu.realtime.common.KafkaConnectorConfig;
import com.atguigu.edu.realtime.function.KafkaDeserWithFilter;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;
import java.util.Properties;

/**
 * @ClassName: KafkaUtil
 * @Description: TODO Kafka工具类,提供kafka消费者和生产者
 * @Author: zhaoxunfeng
 * @Date: 2022-08-31 12:07
 * @Version: 1.0.0
 */
public class KafkaUtil {
    public static FlinkKafkaConsumer<String> getKafkaConsumer(String topic, String groupId) {
        Properties properties = new Properties();
        // 设置kafka连接地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConnectorConfig.KAFKA_BROKER_LIST);
        // 设置 kafka消费者组
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        // 设置kafka消费方式：从最新的位置开始消费
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, KafkaConnectorConfig.OFFSET_STRATEGY);
        // 设置kafka的隔离级别：读已提交
        properties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, KafkaConnectorConfig.READ_COMMITTED);

        return new FlinkKafkaConsumer<String>(topic, new KafkaDeserWithFilter(), properties);
    }

    public static FlinkKafkaProducer<String> getKafkaProducer(String topic) {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConnectorConfig.KAFKA_BROKER_LIST);
        properties.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, KafkaConnectorConfig.TRANSACTION_TIMEOUT);


        return new FlinkKafkaProducer<String>("default_topic",
                new KafkaSerializationSchema<String>() {
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(String value, @Nullable Long aLong) {
                        return new ProducerRecord<>(topic, value.getBytes(StandardCharsets.UTF_8));
                    }
                },
                properties,
                FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }

    //获取kafka连接器相关的连接属性
    public static String getKafkaDDL(String topic, String groupId) {
        String ddl = "WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '" + topic + "',\n" +
                "  'properties.bootstrap.servers' = '" + KafkaConnectorConfig.KAFKA_BROKER_LIST_TEST + "',\n" +
                "  'properties.group.id' = '" + groupId + "',\n" +
                "  'scan.startup.mode' = 'group-offsets',\n" +
                "  'format' = 'json'\n" +
                ")";
        return ddl;
    }

    //获取upsertKafka连接器相关的连接属性
    public static String getUpsertKafkaDDL(String topic) {
        String upsertDDL = "WITH (\n" +
                "  'connector' = 'upsert-kafka',\n" +
                "  'topic' = '" + topic + "',\n" +
                "  'properties.bootstrap.servers' = '" + KafkaConnectorConfig.KAFKA_BROKER_LIST_TEST + "',\n" +
                "  'key.format' = 'json',\n" +
                "  'value.format' = 'json'\n" +
                ")";
        return upsertDDL;
    }

    public static String getKafkaSinkDDL(String topic) {
        return "WITH ( " +
                "  'connector' = 'kafka', " +
                "  'topic' = '" + topic + "', " +
                "  'properties.bootstrap.servers' = '" + KafkaConnectorConfig.KAFKA_BROKER_LIST_TEST + "', " +
                "  'format' = 'json' " +
                ")";

    }
}
